package com.example.flinkcourse.lesson1.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Kafka 数据源工厂类
 * 用于创建 Kafka 数据源
 */
public class KafkaSourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceFactory.class);

    /**
     * 创建 Kafka 数据源
     * @return KafkaSource<String>
     */
    public static KafkaSource<String> createKafkaSource() {
        LOG.info("Creating Kafka source...");

        // 配置 Kafka 属性
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "flink-group");
        
        // 如果启用了 Kerberos 认证
        if (isKerberosEnabled()) {
            configureKerberos(props);
        }

        // 创建 Kafka 数据源
        return KafkaSource.<String>builder()
            .setBootstrapServers(props.getProperty("bootstrap.servers"))
            .setTopics("input-topic-1", "input-topic-2")
            .setGroupId(props.getProperty("group.id"))
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .setProperties(props)
            .build();
    }

    /**
     * 检查是否启用 Kerberos 认证
     */
    private static boolean isKerberosEnabled() {
        // 从配置文件中读取是否启用 Kerberos
        return false; // 默认不启用
    }

    /**
     * 配置 Kerberos 认证
     */
    private static void configureKerberos(Properties props) {
        props.setProperty("security.protocol", "SASL_PLAINTEXT");
        props.setProperty("sasl.mechanism", "GSSAPI");
        props.setProperty("sasl.kerberos.service.name", "kafka");
        props.setProperty("sasl.jaas.config", 
            "org.apache.kafka.common.security.plain.PlainLoginModule required " +
            "username=\"kafka\" password=\"kafka\";");
    }
} 